package com.cn.activeMQ;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.junit.Test;

import javax.jms.*;

/**
 * @program: demo
 * @description: p2p   point to point 点对点模式
 * @author: yihanyuan
 * @create: 2019-08-28 15:07
 */
public class ActiveMqTest {
    @Test
    public void send() throws JMSException {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Topic topic = session.createTopic("topic");
        MessageProducer producer = session.createProducer(topic);
        TextMessage message = session.createTextMessage("message1");
        producer.send(message);
        producer.close();
        session.close();
        connection.close();
    }

    @Test
    public void accept() throws JMSException, InterruptedException {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();
//        .AUTO_ACKNOWLEDGE 自动应答        CLIENT_ACKNOWLEDGE  手动应答
        Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        Topic topic = session.createTopic("topic");
        MessageConsumer consumer = session.createConsumer(topic);
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                try {
                    String text = ((TextMessage) message).getText();
                    if ("message".equals(text)) {
                        System.out.println(((TextMessage) message).getText());
                        message.acknowledge();//手动应答处理了
                    } else {
                        session.recover();//重试
                        System.out.println("失败了");
                        int i = 1 / 0;
                    }
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        Thread.sleep(100000);
    }
}
